package com.ams.protocol.rtmp.net;

import java.io.IOException;
import java.util.LinkedList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.ams.media.IMediaSerializer;
import com.ams.media.MediaMessage;

public class StreamPublisher {
    final private Logger logger = LoggerFactory
            .getLogger(StreamPublisher.class);

    protected String publishName = null;
    private int pingBytes = 0;
    private int lastPing = 0;
    private boolean ping = false;
    protected MediaMessage videoH264Header = null;
    protected MediaMessage audioH264Header = null;
    protected MediaMessage meta = null;
    protected LinkedList<NetStream> subscribers = new LinkedList<NetStream>();

    protected IMediaSerializer recorder = null; // record to file stream

    public StreamPublisher(String publishName) {
        this.publishName = publishName;
        String tokens[] = publishName.split(":");
        if (tokens.length >= 2) {
            this.publishName = tokens[1];
        }
    }

    public void publish(MediaMessage msg) throws IOException {
        if (audioH264Header == null && msg.isH264AudioHeader()) {
            audioH264Header = msg;
            logger.info("received h264 audio header.");
        } else if (videoH264Header == null && msg.isH264VideoHeader()) {
            videoH264Header = msg;
            logger.info("received h264 video header.");
        } else if (msg.isMeta()) {
            meta = msg;
        }

        // ping
        ping(msg.getDataSize());

        // record to file
        record(msg);
        
        // publish packet to subscriber stream
        notify(msg);
    }

    public synchronized void close() {
        if (recorder != null) {
            recorder.close();
        }
        subscribers.clear();
        videoH264Header = null;
        audioH264Header = null;
        meta = null;
    }

    private void ping(int dataSize) {
        pingBytes += dataSize;
        // ping
        ping = false;
        if (pingBytes - lastPing > 1024 * 1) {
            lastPing = pingBytes;
            ping = true;
        }
    }

    private void record(MediaMessage msg) throws IOException {
        if (recorder != null) {
            recorder.write(msg);
        }
    }
    
    private void notify(MediaMessage msg) {
        for (NetStream subscriber : subscribers) {
            try {
                subscriber.sendMediaMessage(msg);
                subscriber.flush();
            } catch (IOException e) {
                removeSubscriber(subscriber);
            }
        }
    }
    
    public void addSubscriber(NetStream subscriber) {
        try {
            if (audioH264Header != null) {
                subscriber.sendMediaMessage(audioH264Header);
            }
            if (videoH264Header != null) {
                subscriber.sendMediaMessage(videoH264Header);
            }
        } catch (IOException e) {}
        synchronized (subscribers) {
            subscribers.add(subscriber);
        }
    }

    public void removeSubscriber(NetStream subscriber) {
        synchronized (subscribers) {
            subscribers.remove(subscriber);
        }
    }

    public void setRecorder(IMediaSerializer recorder) {
        this.recorder = recorder;
    }

    public boolean isPing() {
        return ping;
    }

    public int getPingBytes() {
        return pingBytes;
    }

    public String getPublishName() {
        return publishName;
    }
    
}
